use crate::mq::client::event_loop_task::EventLoopTask;
use rumqttc;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::Duration;
use crate::mq::client;

#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Config {
    pub broker: String,
    pub username: String,
    pub password: String,
}

#[derive(Clone)]
pub struct Setting {
    pub broker: BrokerSetting,
    pub username: String,
    pub password: String,
}

impl TryFrom<&Config> for Setting {
    type Error = anyhow::Error;
    fn try_from(config: &Config) -> Result<Self, Self::Error> {
        let broker_setting = BrokerSetting::try_from(config.broker.as_str())?;
        Ok(Setting {
            broker: broker_setting,
            username: config.username.clone(),
            password: config.password.clone(),
        })
    }
}

#[derive(Clone)]
pub struct BrokerSetting {
    pub transport: rumqttc::Transport,
    pub host: String,
    pub port: u16,
}

impl TryFrom<&str> for BrokerSetting {
    type Error = anyhow::Error;
    fn try_from(broker: &str) -> Result<Self, Self::Error> {
        let broker_url = url::Url::parse(broker).map_err(|e| anyhow::Error::new(e))?;
        let setting = match broker_url.scheme() {
            "tcp" | "mqtt" => BrokerSetting {
                transport: rumqttc::Transport::Tcp,
                host: match broker_url.host() {
                    None => return Err(anyhow::anyhow!("未指定主机")),
                    Some(v) => v.to_string(),
                },
                port: match broker_url.port() {
                    None => return Err(anyhow::anyhow!("未指定端口")),
                    Some(v) => v,
                },
            },
            "ssl" | "mqtts" => BrokerSetting {
                transport: rumqttc::Transport::tls_with_default_config(),
                host: broker.to_string(),
                port: match broker_url.port() {
                    None => return Err(anyhow::anyhow!("未指定端口")),
                    Some(v) => v,
                },
            },
            "ws" => BrokerSetting {
                transport: rumqttc::Transport::ws(),
                host: broker.to_string(),
                port: broker_url.port().unwrap_or_else(|| 80),
            },
            "wss" => BrokerSetting {
                transport: rumqttc::Transport::wss_with_default_config(),
                host: broker.to_string(),
                port: broker_url.port().unwrap_or_else(|| 443),
            },
            _ => return Err(anyhow::anyhow!("不支持的连接类型")),
        };
        Ok(setting)
    }
}

pub struct Client {
    inner: rumqttc::AsyncClient,
    handler_store: Arc<client::event_loop_task::PacketHandlerStore>,
}

impl Client {
    pub fn new(
        setting: Setting,
        client_id: String,
    ) -> Result<(Self, EventLoopTask), anyhow::Error> {
        let mut client_opts =
            rumqttc::MqttOptions::new(client_id, setting.broker.host.as_str(), setting.broker.port);
        client_opts
            .set_credentials(setting.username.as_str(), setting.password.as_str())
            .set_transport(setting.broker.transport.clone())
            .set_keep_alive(Duration::from_secs(20));

        let (client, event_loop) = rumqttc::AsyncClient::new(client_opts, 10);
        let handler_store = Arc::new(client::event_loop_task::PacketHandlerStore::new());
        Ok((
            Client {
                inner: client,
                handler_store: handler_store.clone(),
            },
            EventLoopTask::new(event_loop, handler_store),
        ))
    }

    pub async fn publish_json<T>(&self, topic: String, message: &T) -> Result<(), anyhow::Error>
    where
        T: Serialize,
    {
        let message = serde_json::to_vec(message).map_err(anyhow::Error::new)?;
        self.inner
            .publish(topic, rumqttc::QoS::ExactlyOnce, false, message)
            .await
            .map_err(anyhow::Error::new)?;

        Ok(())
    }

    pub async fn subscribe(
        &self,
        topic: String,
        handler: Arc<dyn client::event_loop_task::PacketHandler>,
    ) -> Result<(), anyhow::Error> {
        self.inner
            .subscribe(topic.clone(), rumqttc::QoS::ExactlyOnce)
            .await
            .map_err(anyhow::Error::new)?;

        self.handler_store.store(topic, handler);
        Ok(())
    }

    pub async fn unsubscribe(&self, topic: String) -> Result<(), anyhow::Error> {
        self.inner
            .unsubscribe(topic.clone())
            .await
            .map_err(anyhow::Error::new)?;

        self.handler_store.delete(topic);
        Ok(())
    }
}

pub struct Factory {
    setting: Setting,
}

impl Factory {
    pub fn new(setting: Setting) -> Self {
        Factory { setting }
    }

    pub fn create(&self) -> Result<(Client, EventLoopTask), anyhow::Error> {
        let client_id = wanmesh_common::util::new_id()?;
        Client::new(self.setting.clone(), client_id)
    }
}
